package com.ben.chatroom.distribute;

import com.ben.chatroom.common.constants.ServerConstants;
import com.ben.chatroom.common.entity.ImNode;
import com.ben.chatroom.common.util.ObjectUtil;
import com.ben.chatroom.common.util.ThreadUtils;
import com.ben.chatroom.common.zk.CuratorZKclient;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * @author huangyijun
 */
@Slf4j
public class WorkerRouter {


  private CuratorFramework client;
  //是否初始化
  private boolean inited = false;
  private static WorkerRouter singleInstance;
  private static final String path = ServerConstants.MAGIC_PATH;

  private final ConcurrentHashMap<Long, PeerSender> workerMap = new ConcurrentHashMap<>();

  private WorkerRouter() {
  }

  public synchronized static WorkerRouter getInst() {
    if(null == singleInstance) {
      singleInstance = new WorkerRouter();
    }
    return singleInstance;
  }

  private BiConsumer<ImNode, PeerSender> runAfterAdd = (node, relaySender) -> {
    if(null != relaySender) {
      //关闭老的连接
      relaySender.stopConnecting();
    }
    relaySender = new PeerSender(node);
    //建立转发的连接
    relaySender.doConnect();

    workerMap.put(node.getId(), relaySender);
  };

  private Consumer<ImNode> runAfterRemove = (node) -> {
    if(workerMap.containsKey(node.getId())) {
      PeerSender peerSender = workerMap.get(node.getId());
      peerSender.stopConnecting();
      workerMap.remove(node.getId());
    }
  };

  /**
   * 初始化节点管理
   */
  public void init() {
    if(inited) {
      return;
    }

    if(null == client) {
      this.client = CuratorZKclient.getSingleton().getClient();
    }

    try {
      //订阅节点的增加和删除事件
      PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);
      PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
          log.info("开始监听其他ImWorker字节点：-------");
          ChildData data = event.getData();
          switch(event.getType()) {
            case CHILD_ADDED:
              log.info("CHILD_ADDED : " + data.getPath() + "  数据:" + Arrays.toString(data.getData()));
              processNodeAdd(data);
              break;
            case CHILD_UPDATED:
              log.info("CHILD_UPDATED : " + data.getPath() + "  数据:" + new String(data.getData()));
              break;
            case CHILD_REMOVED:
              log.info("CHILD_REMOVED : " + data.getPath() + "  数据:" + Arrays.toString(data.getData()));
              processNodeRemoved(data);
              break;
            default:
              log.debug("[PathChildrenCache]节点数据为空, path={}", data == null ? "null" : data.getPath());
              break;
          }

        }
      };
      childrenCache.getListenable().addListener(childrenCacheListener, ThreadUtils.getIoIntenseTargetThreadPool());
      log.info("注册zk监控器成功!");
      inited = true;
      childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
    }
    catch(Exception e) {
      log.error(e.getMessage());
    }
  }

  /**
   * 节点增加
   */
  private void processNodeAdd(ChildData data) {
    byte[] payload = data.getData();
    ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);

    long id = ImWoker.getInst().getIdByPath(data.getPath());
    node.setId(id);

    log.info("[TreeCache]节点更新端口，path={},data={}", data.getPath(), ObjectUtil.ObjectToJson(node));
    if(node.equals(getLocalNode())) {
      log.info("[TreeCache]本地节点, path={}, data={}",
        data.getPath(), ObjectUtil.ObjectToJson(node));
      return;
    }
    PeerSender peerSender = workerMap.get(id);
    //重复收到注册的事件
    if(null != peerSender && peerSender.getRmNode().equals(node)) {
      log.info("[TreeCache]节点重复增加,path={},data={}", data.getPath(), node);
      return;
    }

    if(runAfterAdd != null) {
      runAfterAdd.accept(node, peerSender);
    }

  }

  /**
   * 节点减少
   */
  private void processNodeRemoved(ChildData data) {
    byte[] payload = data.getData();
    ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);

    long id = ImWoker.getInst().getIdByPath(data.getPath());
    node.setId(id);
    log.info("TreeCach节点删除,path={},data={}", data.getPath(), node);

    if(runAfterRemove != null) {
      runAfterRemove.accept(node);
    }

  }

  public ImNode getLocalNode() {
    return ImWoker.getInst().getLocalNode();
  }

  /**
   * 根据结点id查询
   * @param nodeId 节点id
   * @return
   */
  public PeerSender router(long nodeId) {
    return workerMap.get(nodeId);
  }
}
